package cn.mjnxin.iot.gate.handle;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;

import cn.mjnxin.iot.base.domain.ChannelData;
import cn.mjnxin.iot.base.domain.TerminalRepacker;
import cn.mjnxin.iot.base.log.LogWorker;
import cn.mjnxin.iot.base.monitor.CountMonitor;
import cn.mjnxin.iot.base.util.CommonUtil;
import cn.mjnxin.iot.gate.cache.GateRuntimeCache;
import cn.mjnxin.iot.gate.util.GateUtil;
import cn.mjnxin.iot.gate.worker.Gate2HostWorker;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * 接收前置报文转发终端
 * @author CLG 2023年10月10日
 * @version V1.0.0
 */
@Sharable
public class GateAsClient2TerminalInHandler extends ChannelInboundHandlerAdapter{
    private final Integer pid;
	// 日志等级 0：无日志 1：报文长度日志 2：报文内容日志 3：所有日志
	private final Integer logLevel;
    private final TerminalRepacker repacker;
    
    public GateAsClient2TerminalInHandler(Integer pid, Integer logLevel, TerminalRepacker repacker) {
        this.pid = pid;
        this.logLevel = logLevel;
        this.repacker = repacker;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof List) {
            @SuppressWarnings("unchecked")
            List<ChannelData> lists = (List<ChannelData>) msg;

            for (ChannelData channelData : lists) {
            	Gate2HostWorker.getInstance().add(channelData);
                analyse();
            }

        } else {
            ChannelData channelData = (ChannelData) msg;
            Gate2HostWorker.getInstance().add(channelData);
            analyse();
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
//        super.channelActive(ctx);
        Channel channel = ctx.channel();
        InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
        if (insocket != null) {
            String terminalIpAddress = CommonUtil.getIpAddress(insocket);
            GateRuntimeCache.terminalLink(pid, terminalIpAddress, channel);
            Integer count = GateRuntimeCache.getCount(terminalIpAddress);
            // 每个终端一套count，short够用了
            if (count != null && count.intValue() < 0x7FFF) {
                GateRuntimeCache.setCount(terminalIpAddress, count + 1);
            } else {
                GateRuntimeCache.setCount(terminalIpAddress, 1);
            }
            LogWorker.logForce(String.format("%d|Add Terminal[%s]", pid, terminalIpAddress));
            GateUtil.terminalLink(pid, logLevel, terminalIpAddress, GateRuntimeCache.gateNo);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        Channel channel = ctx.channel();
        InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
        if (insocket != null) {
        	System.err.println("GateAsClient2Terminal.channelInactive(ChannelHandlerContext ctx),Adress:" + insocket);
            /**
             * Terminal主动断连时删除缓存中对应的client的缓存
             */
            String ipAddress = CommonUtil.getIpAddress(insocket);
            GateRuntimeCache.terminalUnlink(ipAddress);
            GateUtil.terminalUnlink(pid, logLevel, ipAddress, GateRuntimeCache.gateNo);
            if (repacker != null) {
            	repacker.unlinkTerminal(ipAddress);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
    	System.err.println("GateAsClient2Terminal.exceptionCaught(ChannelHandlerContext ctx, Throwable cause),Adress:" + insocket);
        if (cause instanceof IOException || cause instanceof OutOfMemoryError) {
            LogWorker.logForce(String.format("%d|Terminal[%s]断连", pid, insocket.toString()));
        } else {
            LogWorker.logForce(String.format("%d|Terminal[%s]其他异常", pid, insocket.toString()));
        }
        super.exceptionCaught(ctx, cause);
    }

    /**
     * 心跳检测触发的事件通过本方法捕获
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    	System.err.println("GateAsClient2Terminal.userEventTriggered(ChannelHandlerContext ctx, Object evt) " + evt);
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            // 心跳检测,超时时间xx秒，指定时间中没有读写操作会触发IdleStateEvent事件
            if (state == IdleState.READER_IDLE) {
                ChannelFuture fcutrue = ctx.close();
                fcutrue.addListener(new ChannelFutureListener() {
                    public void operationComplete(ChannelFuture future) throws Exception {
                        Channel channel = future.channel();
                        InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
                        if (insocket != null) {
                        	System.err.println("GateAsClient2Terminal.userEventTriggered(ChannelHandlerContext ctx, Object evt),Adress:" + insocket);
	                        String ipAddress = CommonUtil.getIpAddress(insocket);
	                        LogWorker.logForce(String.format("%d|Terminal[%s]心跳超时下线成功....", pid, ipAddress));
                        }
                        channel.close();
                    }
                });
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    /**
     * 统计每1000次并发耗时 单位ms
     */
    private void analyse() {
        if (CountMonitor.startTimeLong.get() == 0) {
            synchronized (CountMonitor.class) {
                if (CountMonitor.startTimeLong.get() == 0) {
                    CountMonitor.startTimeLong.set(System.currentTimeMillis());
                }
            }
        }
        CountMonitor.terminalRecieveCount.addAndGet(1);
        long curNum = CountMonitor.terminalRecieveCount.get();
        if (curNum % 1000 == 0) {
            LogWorker.logForce(curNum + "耗时：" + (System.currentTimeMillis() - CountMonitor.startTimeLong.get()));
        }
    }
}
